package amqpclt

import (
	"context"
	carpb "coolcar/v1/car/api/gen/v1"
	"encoding/json"
	"fmt"
	"github.com/streadway/amqp"
	"go.uber.org/zap"
)

type Publisher struct {
	ch       *amqp.Channel
	exchange string
}

func NewPublisher(conn *amqp.Connection, exchange string) (*Publisher, error) {
	ch, err := conn.Channel()
	if err != nil {
		return nil, fmt.Errorf("failed to create channel: %v", err)
	}
	err = declareExchange(ch, exchange)
	if err != nil {
		return nil, fmt.Errorf("failed to declare channel: %v", err)
	}
	return &Publisher{
		ch:       ch,
		exchange: exchange,
	}, nil
}

func (p Publisher) Publish(ctx context.Context, car *carpb.CarEntity) error {
	bytes, err := json.Marshal(car)
	if err != nil {
		return fmt.Errorf("failed to marshal car: %v", err)
	}
	return p.ch.Publish(
		p.exchange, "", false, false, amqp.Publishing{
			Body: bytes,
		})
}

type Subscriber struct {
	conn     *amqp.Connection
	exchange string
	Sugar    *zap.SugaredLogger
}

func NewSubscriber(conn *amqp.Connection, exchange string, Sugar *zap.SugaredLogger) (*Subscriber, error) {
	ch, err := conn.Channel()
	if err != nil {
		return nil, fmt.Errorf("failed to create channel: %v", err)
	}
	err = declareExchange(ch, exchange)
	if err != nil {
		return nil, fmt.Errorf("failed to declare channel: %v", err)
	}

	defer func(ch *amqp.Channel) {
		err := ch.Close()
		if err != nil {
			Sugar.Errorf("failed to close channel: %v", err)
		}
	}(ch)
	return &Subscriber{
		conn:     conn,
		exchange: exchange,
		Sugar:    Sugar,
	}, nil
}

func (s *Subscriber) SubscribeRaw(ctx context.Context) (<-chan amqp.Delivery, func(), error) {
	ch, err := s.conn.Channel()
	if err != nil {
		return nil, func() {}, fmt.Errorf("failed to create channel: %v", err)
	}

	queue, err := ch.QueueDeclare("", false, true, false, false, nil)
	closeCh := func() {
		err := ch.Close()
		if err != nil {
			s.Sugar.Errorf("failed to close channel: %v", err)
		}
	}
	if err != nil {
		return nil, closeCh, fmt.Errorf("failed to declare channel: %v", err)
	}

	cleanUp := func() {
		_, err := ch.QueueDelete(queue.Name, false, false, false)
		if err != nil {
			s.Sugar.Errorf("failed to delete channel: %v", err)
		}
		closeCh()
	}
	err = ch.QueueBind(queue.Name, "", s.exchange, false, nil)
	if err != nil {
		return nil, cleanUp, fmt.Errorf("failed to bind channel: %v", err)
	}

	msgs, err := ch.Consume(queue.Name, "", true, false, false, false, nil)
	if err != nil {
		return nil, cleanUp, fmt.Errorf("failed to consume channel: %v", err)
	}
	return msgs, cleanUp, nil
}

func (s *Subscriber) Subscribe(ctx context.Context) (chan *carpb.CarEntity, func(), error) {
	msgCh, cleanUp, err := s.SubscribeRaw(ctx)
	if err != nil {
		return nil, cleanUp, err
	}
	carCh := make(chan *carpb.CarEntity)
	go func() {
		for msg := range msgCh {
			var car carpb.CarEntity
			err := json.Unmarshal(msg.Body, &car)
			if err != nil {
				s.Sugar.Errorf("failed to unmarshal: %v", err)
			}
			carCh <- &car
		}
		close(carCh)
	}()
	return carCh, cleanUp, nil
}

func declareExchange(ch *amqp.Channel, exchange string) error {
	return ch.ExchangeDeclare(exchange, "fanout", true, false, false, false, nil)
}
